"""
Case Type   : 协议兼容
Case Name   : sha2协议连接成功，database相关sql验证操作
Create At   : 2024.09.23
Owner       : @zhanghuan96
Description :
    1、创建兼容b库并查询插件是否加载成功
    2、创建用户
    3、caching_sha2认证
    4、查询系统未使用端口
    5、设置协议兼容guc参数
    6、连接og侧B库创建database
    7、连接og侧B库创建database添加charset、COLLAT
    8、连接og侧B库创建database添加添加encoding
    9、连接og侧B库修改database
    10、连接og侧B库修改databas
    11、清理环境
Expect      :
    1、成功，插件查询成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功，实际为create schema
    7、成功，实际为create schema
    8、成功，实际为create database
    9、成功，实际为alter schema
    10、成功，实际为alter database
    11、清理成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonMySQL import CommonMySQL
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class CompatibilityTest01(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_pri = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.my_node = Node('mysql8')
        self.constant = Constant()
        self.db_name = 'db_protocol_0003'
        self.u_name = 'u_protocol_0003'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      'postgresql.conf')
        self.default_value1 = self.common.show_param('enable_dolphin_proto')
        self.default_value2 = self.common.show_param('dolphin_server_port')
        self.exec_sql1 = (f'set dolphin.b_compatibility_mode = on;'
                          f'drop database if exists {self.db_name}_01;'
                          f'create database {self.db_name}_01;'
                          f'show create database {self.db_name}_01;')
        self.exec_sql2 = (f"set dolphin.b_compatibility_mode = on;"
                          f"drop database if exists {self.db_name}_02;"
                          f"create database {self.db_name}_02  charset "
                          f"'SQL_ASCII' COLLATE='SQL_ASCII';"
                          f"show create database {self.db_name}_02;")
        self.exec_sql3 = (f"set dolphin.b_compatibility_mode = on;"
                          f"drop database if exists {self.db_name}_03;"
                          f"create database {self.db_name}_03 encoding 'utf8';"
                          f"select datname from pg_database where "
                          f"datname='{self.db_name}_03';")
        self.exec_sql4 = (f"set dolphin.b_compatibility_mode = on;"
                          f" alter database {self.db_name}_01  "
                          f"charset 'SQL_ASCII';")
        self.exec_sql5 = (f"set dolphin.b_compatibility_mode = on;"
                          f"alter database {self.db_name}_03 rename to "
                          f"{self.db_name}_03_new;"
                          f"select datname from pg_database where "
                          f" datname='{self.db_name}_03_new';")

    def test_create_function(self):
        text = '-----step1:创建兼容b库并查询插件;expect:成功-----'
        self.log.info(text)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name}  dbcompatibility ='B';")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertEqual(1,
                         sql_res.count(self.constant.CREATE_DATABASE_SUCCESS),
                         '兼容B库创建失败')
        sql_cmd1 = "select extname from pg_extension where extname ='dolphin';"
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('dolphin', sql_res1, '查询失败')

        text = '---step2:创建用户;expect:成功---'
        self.log.info(text)
        sql_cmd2 = (f"drop user if exists {self.u_name};"
                    f"create user {self.u_name} with password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"grant all privileges to {self.u_name};")
        self.log.info(sql_cmd2)
        sql_res2 = self.sh_pri.execut_db_sql(sql_cmd2, dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('CREATE ROLE', sql_res2, '创建失败')
        self.assertIn('ALTER ROLE', sql_res2, '创建失败')

        text = '---step3:caching_sha2认证;expect:成功---'
        self.log.info(text)
        sql_cmd = (f"select set_caching_sha2_password('{self.u_name}', "
                   f"'{macro.COMMON_PASSWD}', '');")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(1 row)', sql_res, '创建失败')

        text = '----step4:查询系统未使用端口;expect:成功----'
        self.log.info(text)
        port = self.common.get_not_used_port(self.pri_node)
        self.assertNotEqual(0, port, '执行失败:' + text)

        text = '-step5:设置协议兼容guc参数;expect:成功-'
        self.log.info(text)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_dolphin_proto = on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'dolphin_server_port ={port}')
        self.log.info(msg)
        self.assertTrue(msg)
        sed_cmd = (f"sed -i '$a dolphin.default_database_name={self.db_name}' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-step6:连接og侧B库创建database;expect:成功，实际为create schema--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql1, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn(f'CREATE SCHEMA {self.db_name}_01 AUTHORIZATION '
                      'u_protocol_0003', c_res)

        text = ('-step7:连接og侧B库创建database添加charset、COLLAT;'
                'expect:成功，实际为create schema--')
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql2, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn(f'CREATE SCHEMA {self.db_name}_02 '
                      'AUTHORIZATION u_protocol_0003', c_res)

        text = ('-step8:连接og侧B库创建database添加添加encoding;'
                'expect:成功，实际为create database--')
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql3, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)
        self.assertIn(f'{self.db_name}_03', c_res)

        text = '-step9:连接og侧B库修改database;expect:成功，实际为alter schema--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql4, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step10:连接og侧B库修改database;expect:成功，实际为alter database--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql5, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn(f'{self.db_name}_03_new', c_res)

    def tearDown(self):
        text = '-----step11:清理环境;expect:成功-----'
        self.log.info(text)
        dr_cmd1 = (f"drop schema if exists {self.db_name}_01;"
                   f"drop schema if exists {self.db_name}_02;")
        self.log.info(dr_cmd1)
        dr_res1 = self.sh_pri.execut_db_sql(dr_cmd1,
                                            sql_type=f'-U {self.u_name} '
                                                     f'-W{macro.COMMON_PASSWD}',
                                            dbname=self.db_name)
        self.log.info(dr_res1)
        drop_cmd = (f"drop database if exists {self.db_name};"
                    f"drop database if exists {self.db_name}_03_new;")
        self.log.info(drop_cmd)
        drop_res = self.sh_pri.execut_db_sql(drop_cmd, dbname='postgres')
        self.log.info(drop_res)
        drop_cmd1 = f"drop user if exists {self.u_name} cascade;"
        self.log.info(drop_cmd1)
        drop_res1 = self.sh_pri.execut_db_sql(drop_cmd1)
        self.log.info(drop_res1)

        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'enable_dolphin_proto = '
                                            f'{self.default_value1}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ='
                                            f'{self.default_value2}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)

        sed_cmd = (f"sed -i '/dolphin.default_database_name={self.db_name}/d' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertEqual(2, dr_res1.count('DROP SCHEMA'), '清理环境失败')
        self.assertIn('DROP ROLE', drop_res1, '清理环境失败')
        self.assertEqual(2,
                         drop_res.count(self.constant.DROP_DATABASE_SUCCESS),
                         '清理环境失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
